package com.atbeijing.D04;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 自定义水位线
 */
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .readTextFile("C:\\Users\\WangJinGen\\IdeaProjects\\flink\\data\\UserBehavior.csv")
                .map(new MapFunction<String, Example5.UserBehavior>() {
                    @Override
                    public Example5.UserBehavior map(String value) throws Exception {
                        String[] arr = value.split(",");
                        return new Example5.UserBehavior(
                                arr[0],arr[1],arr[2],arr[3],
                                Long.parseLong(arr[4]) * 1000L
                        );
                    }
                })
                .filter(r -> r.behaviorType.equals("pv"))
                .assignTimestampsAndWatermarks(new CustomWatermarkStrategy())
                .print();
        env.execute();
    }

    //泛型为事件的类型
    public static class CustomWatermarkStrategy implements WatermarkStrategy<Example5.UserBehavior>{

        @Override
        public WatermarkGenerator<Example5.UserBehavior> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            return new CustomBoundedOutOfOrdernessAssigner();
        }

        @Override
        public TimestampAssigner<Example5.UserBehavior> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new SerializableTimestampAssigner<Example5.UserBehavior>() {
                @Override
                public long extractTimestamp(Example5.UserBehavior element, long recordTimestamp) {
                    return element.timestamp; // 告诉程序数据源里的时间戳是哪一个字段
                }
            };
        }
    }

    public static class CustomBoundedOutOfOrdernessAssigner implements WatermarkGenerator<Example5.UserBehavior> {
        private Long delayTime = 5000L; // 最大延迟时间
        private Long maxTs = -Long.MAX_VALUE + delayTime + 1L; // 初始化观察到的最大时间戳
        @Override
        public void onEvent(Example5.UserBehavior event, long eventTimestamp, WatermarkOutput output) {
            // 每来一条数据就调用一次
            maxTs = Math.max(event.timestamp, maxTs); // 更新最大时间戳
        }

        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            // 发射水位线,也是改表逻辑时钟
            // 默认200ms调用一次
            output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
        }
    }
}
